package cn.gwm.flink.streaming.dws.reservecharge;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.dwm.charge.util.SupplyUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.util.ObjectUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.math.BigDecimal;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author Liu.Nanfang
 * @date 2022/12/26 14:53
 * @description 异步处理添加gps
 */
public class HttpAsyncGpsFunction extends RichAsyncFunction<ReserveChargeData, ReserveChargeData> {
    private static final long serialVersionUID = 8522411971886428444L;

    private static final long TERMINATION_TIMEOUT = 5000L;
    private static final int THREAD_POOL_SIZE = 20;
    static ExecutorService executorService;
    static int counter = 0;

    static String url;

    @Override
    public void open(Configuration parameters) throws Exception {
        synchronized (HttpAsyncGpsFunction.class) {
            if (counter == 0) {
                executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
            }
            ++counter;
        }
        url = ConfigLoader.get("charging.station.type.url");
    }

    private void freeExecutor() {
        synchronized (HttpAsyncGpsFunction.class) {
            --counter;
            if (counter == 0) {
                executorService.shutdown();

                try {
                    if (!executorService.awaitTermination(TERMINATION_TIMEOUT, TimeUnit.MILLISECONDS)) {
                        executorService.shutdownNow();
                    }
                } catch (InterruptedException interrupted) {
                    executorService.shutdownNow();

                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        freeExecutor();
    }

    @Override
    public void asyncInvoke(ReserveChargeData in, ResultFuture<ReserveChargeData> resultFuture) throws Exception {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                ReserveChargeData result = queryGpsByWin(in);
                resultFuture.complete(Collections.singleton(result));
            }
        });
    }

    /**
     * 异步获取经纬度信息
     *
     * @param in
     * @return
     */
    public ReserveChargeData queryGpsByWin(ReserveChargeData in) {
        if (ObjectUtil.isNotNull(in.getAdCode()) && !"[]".equals(in.getAdCode()) && !BaseFields.defaultVal.equals(in.getAdCode())) {
            return in;
        }
        if (ObjectUtil.isNotNull(in.getLatitude()) && ObjectUtil.isNotNull(in.getLongitude())) {
            if (!BaseFields.defaultVal.equals(in.getLatitude().toString()) &&
                    !BaseFields.defaultVal.equals(in.getLongitude().toString())) {
                return in;
            }
        }

        Map<String, BigDecimal> gps = SupplyUtil.getGpsByVinAndTid(in.getDeviceId(), in.getItem_time() - 5000L, in.getItem_time() + 5000L);
        if (ObjectUtil.isNotEmpty(gps)) {
            in.setLatitude(gps.get("latitude"));
            in.setLongitude(gps.get("longitude"));
        }


        return in;

    }
}

